/*
 * Copyright 2016 Netflix, Inc.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.netflix.conductor.common.metadata.tasks;

import com.github.vmg.protogen.annotations.ProtoEnum;
import com.github.vmg.protogen.annotations.ProtoField;
import com.github.vmg.protogen.annotations.ProtoMessage;
import com.google.protobuf.Any;
import com.netflix.conductor.common.metadata.workflow.TaskType;
import com.netflix.conductor.common.metadata.workflow.WorkflowTask;

import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@ProtoMessage
public class Task {

    @ProtoEnum
    public enum Status {

        IN_PROGRESS(false, true, true),
        CANCELED(true, false, false),
        FAILED(true, false, true),
        FAILED_WITH_TERMINAL_ERROR(true, false, false), //No Retires even if retries are configured, the task and the related workflow should be terminated
        COMPLETED(true, true, true),
        COMPLETED_WITH_ERRORS(true, true, true),
        SCHEDULED(false, true, true),
        TIMED_OUT(true, false, true),
        READY_FOR_RERUN(false, true, true),
        SKIPPED(true, true, false);

        private boolean terminal;

        private boolean successful;

        private boolean retriable;

        Status(boolean terminal, boolean successful, boolean retriable) {
            this.terminal = terminal;
            this.successful = successful;
            this.retriable = retriable;
        }

        public boolean isTerminal() {
            return terminal;
        }

        public boolean isSuccessful() {
            return successful;
        }

        public boolean isRetriable() {
            return retriable;
        }
    }

    @ProtoField(id = 1)
    private String taskType;

    @ProtoField(id = 2)
    private Status status;

    @ProtoField(id = 3)
    private Map<String, Object> inputData = new HashMap<>();

    @ProtoField(id = 4)
    private String referenceTaskName;

    @ProtoField(id = 5)
    private int retryCount;

    @ProtoField(id = 6)
    private int seq;

    @ProtoField(id = 7)
    private String correlationId;

    @ProtoField(id = 8)
    private int pollCount;

    @ProtoField(id = 9)
    private String taskDefName;

    /**
     * Time when the task was scheduled
     */
    @ProtoField(id = 10)
    private long scheduledTime;

    /**
     * Time when the task was first polled
     */
    @ProtoField(id = 11)
    private long startTime;

    /**
     * Time when the task completed executing
     */
    @ProtoField(id = 12)
    private long endTime;

    /**
     * Time when the task was last updated
     */
    @ProtoField(id = 13)
    private long updateTime;

    @ProtoField(id = 14)
    private int startDelayInSeconds;

    @ProtoField(id = 15)
    private String retriedTaskId;

    @ProtoField(id = 16)
    private boolean retried;

    @ProtoField(id = 17)
    private boolean executed;

    @ProtoField(id = 18)
    private boolean callbackFromWorker = true;

    @ProtoField(id = 19)
    private long responseTimeoutSeconds;

    @ProtoField(id = 20)
    private String workflowInstanceId;

    @ProtoField(id = 21)
    private String workflowType;

    @ProtoField(id = 22)
    private String taskId;

    @ProtoField(id = 23)
    private String reasonForIncompletion;

    @ProtoField(id = 24)
    private long callbackAfterSeconds;

    @ProtoField(id = 25)
    private String workerId;

    @ProtoField(id = 26)
    private Map<String, Object> outputData = new HashMap<>();

    @ProtoField(id = 27)
    private WorkflowTask workflowTask;

    @ProtoField(id = 28)
    private String domain;

    @ProtoField(id = 29)
    private Any inputMessage;

    @ProtoField(id = 30)
    private Any outputMessage;

    // This field is deprecated, do not reuse id 31.
    //@ProtoField(id = 31)
    //private int rateLimitPerSecond;

    @ProtoField(id = 32)
    private int rateLimitPerFrequency;

    @ProtoField(id = 33)
    private int rateLimitFrequencyInSeconds;

    @ProtoField(id = 34)
    private String externalInputPayloadStoragePath;

    @ProtoField(id = 35)
    private String externalOutputPayloadStoragePath;

    public Task() {
    }

    /**
     * @return Type of the task
     * @see TaskType
     */
    public String getTaskType() {
        return taskType;
    }

    public void setTaskType(String taskType) {
        this.taskType = taskType;
    }

    /**
     * @return Status of the task
     */
    public Status getStatus() {
        return status;
    }

    /**
     * @param status Status of the task
     */
    public void setStatus(Status status) {
        this.status = status;
    }

    @Deprecated
    public Status getTaskStatus() {
        return status;
    }

    @Deprecated
    public void setTaskStatus(Status taskStatus) {
        this.status = taskStatus;
    }

    public Map<String, Object> getInputData() {
        return inputData;
    }

    public void setInputData(Map<String, Object> inputData) {
        this.inputData = inputData;
    }


    /**
     * @return the referenceTaskName
     */
    public String getReferenceTaskName() {
        return referenceTaskName;
    }

    /**
     * @param referenceTaskName the referenceTaskName to set
     */
    public void setReferenceTaskName(String referenceTaskName) {
        this.referenceTaskName = referenceTaskName;
    }

    /**
     * @return the correlationId
     */
    public String getCorrelationId() {
        return correlationId;
    }

    /**
     * @param correlationId the correlationId to set
     */
    public void setCorrelationId(String correlationId) {
        this.correlationId = correlationId;
    }

    /**
     * @return the retryCount
     */
    public int getRetryCount() {
        return retryCount;
    }

    /**
     * @param retryCount the retryCount to set
     */
    public void setRetryCount(int retryCount) {
        this.retryCount = retryCount;
    }

    /**
     * @return the scheduledTime
     */
    public long getScheduledTime() {
        return scheduledTime;
    }

    /**
     * @param scheduledTime the scheduledTime to set
     */
    public void setScheduledTime(long scheduledTime) {
        this.scheduledTime = scheduledTime;
    }

    /**
     * @return the startTime
     */
    public long getStartTime() {
        return startTime;
    }

    /**
     * @param startTime the startTime to set
     */
    public void setStartTime(long startTime) {
        this.startTime = startTime;
    }

    /**
     * @return the endTime
     */
    public long getEndTime() {
        return endTime;
    }

    /**
     * @param endTime the endTime to set
     */
    public void setEndTime(long endTime) {
        this.endTime = endTime;
    }


    /**
     * @return the startDelayInSeconds
     */
    public int getStartDelayInSeconds() {
        return startDelayInSeconds;
    }

    /**
     * @param startDelayInSeconds the startDelayInSeconds to set
     */
    public void setStartDelayInSeconds(int startDelayInSeconds) {
        this.startDelayInSeconds = startDelayInSeconds;
    }

    /**
     * @return the retriedTaskId
     */
    public String getRetriedTaskId() {
        return retriedTaskId;
    }

    /**
     * @param retriedTaskId the retriedTaskId to set
     */
    public void setRetriedTaskId(String retriedTaskId) {
        this.retriedTaskId = retriedTaskId;
    }

    /**
     * @return the seq
     */
    public int getSeq() {
        return seq;
    }

    /**
     * @param seq the seq to set
     */
    public void setSeq(int seq) {
        this.seq = seq;
    }

    /**
     * @return the updateTime
     */
    public long getUpdateTime() {
        return updateTime;
    }

    /**
     * @param updateTime the updateTime to set
     */
    public void setUpdateTime(long updateTime) {
        this.updateTime = updateTime;
    }


    /**
     * @return the queueWaitTime
     */
    public long getQueueWaitTime() {
        if (this.startTime > 0 && this.scheduledTime > 0) {
            return this.startTime - scheduledTime - (getCallbackAfterSeconds() * 1000);
        }
        return 0L;
    }

    public void setQueueWaitTime(long t) {

    }

    /**
     * @return True if the task has been retried after failure
     */
    public boolean isRetried() {
        return retried;
    }

    /**
     * @param retried the retried to set
     */
    public void setRetried(boolean retried) {
        this.retried = retried;
    }

    /**
     * @return True if the task has completed its lifecycle within conductor (from start to completion to being updated in the datastore)
     */
    public boolean isExecuted() {
        return executed;
    }

    /**
     * @param executed the executed value to set
     */
    public void setExecuted(boolean executed) {
        this.executed = executed;
    }

    /**
     * @return No. of times task has been polled
     */
    public int getPollCount() {
        return pollCount;
    }

    public void setPollCount(int pollCount) {
        this.pollCount = pollCount;
    }


    public boolean isCallbackFromWorker() {
        return callbackFromWorker;
    }

    public void setCallbackFromWorker(boolean callbackFromWorker) {
        this.callbackFromWorker = callbackFromWorker;
    }

    /**
     * @return Name of the task definition
     */
    public String getTaskDefName() {
        if (taskDefName == null || "".equals(taskDefName)) {
            taskDefName = taskType;
        }
        return taskDefName;
    }

    /**
     * @param taskDefName Name of the task definition
     */
    public void setTaskDefName(String taskDefName) {
        this.taskDefName = taskDefName;
    }


    /**
     * @return the timeout for task to send response.  After this timeout, the task will be re-queued
     */
    public long getResponseTimeoutSeconds() {
        return responseTimeoutSeconds;
    }

    /**
     * @param responseTimeoutSeconds - timeout for task to send response.  After this timeout, the task will be re-queued
     */
    public void setResponseTimeoutSeconds(long responseTimeoutSeconds) {
        this.responseTimeoutSeconds = responseTimeoutSeconds;
    }


    /**
     * @return the workflowInstanceId
     */
    public String getWorkflowInstanceId() {
        return workflowInstanceId;
    }

    /**
     * @param workflowInstanceId the workflowInstanceId to set
     */
    public void setWorkflowInstanceId(String workflowInstanceId) {
        this.workflowInstanceId = workflowInstanceId;
    }

    public String getWorkflowType() {
        return workflowType;
    }


    /**
     * @param workflowType the name of the workflow
     * @return the task object with the workflow type set
     */
    public Task setWorkflowType(String workflowType) {
        this.workflowType = workflowType;
        return this;
    }

    /**
     * @return the taskId
     */
    public String getTaskId() {
        return taskId;
    }

    /**
     * @param taskId the taskId to set
     */
    public void setTaskId(String taskId) {
        this.taskId = taskId;
    }

    /**
     * @return the reasonForIncompletion
     */
    public String getReasonForIncompletion() {
        return reasonForIncompletion;
    }

    /**
     * @param reasonForIncompletion the reasonForIncompletion to set
     */
    public void setReasonForIncompletion(String reasonForIncompletion) {
        this.reasonForIncompletion = reasonForIncompletion;
    }

    /**
     * @return the callbackAfterSeconds
     */
    public long getCallbackAfterSeconds() {
        return callbackAfterSeconds;
    }

    /**
     * @param callbackAfterSeconds the callbackAfterSeconds to set
     */
    public void setCallbackAfterSeconds(long callbackAfterSeconds) {
        this.callbackAfterSeconds = callbackAfterSeconds;
    }

    /**
     * @return the workerId
     */
    public String getWorkerId() {
        return workerId;
    }

    /**
     * @param workerId the workerId to set
     */
    public void setWorkerId(String workerId) {
        this.workerId = workerId;
    }

    /**
     * @return the outputData
     */
    public Map<String, Object> getOutputData() {
        return outputData;
    }

    /**
     * @param outputData the outputData to set
     */
    public void setOutputData(Map<String, Object> outputData) {
        this.outputData = outputData;
    }

    /**
     * @return Workflow Task definition
     */
    public WorkflowTask getWorkflowTask() {
        return workflowTask;
    }

    /**
     * @param workflowTask Task definition
     */
    public void setWorkflowTask(WorkflowTask workflowTask) {
        this.workflowTask = workflowTask;
    }

    /**
     * @return the domain
     */
    public String getDomain() {
        return domain;
    }

    /**
     * @param domain the Domain
     */
    public void setDomain(String domain) {
        this.domain = domain;
    }

    public Any getInputMessage() {
        return inputMessage;
    }

    public void setInputMessage(Any inputMessage) {
        this.inputMessage = inputMessage;
    }

    public void setRateLimitPerFrequency(int rateLimitPerFrequency) {
        this.rateLimitPerFrequency = rateLimitPerFrequency;
    }

    public Any getOutputMessage() {
        return outputMessage;
    }

    public void setOutputMessage(Any outputMessage) {
        this.outputMessage = outputMessage;
    }

    /**
     * @return {@link Optional} containing the task definition if available
     */
    public Optional<TaskDef> getTaskDefinition() {
        return Optional.ofNullable(this.getWorkflowTask())
                .map(WorkflowTask::getTaskDefinition);
    }

    public int getRateLimitPerFrequency() {
        return rateLimitPerFrequency;
    }

    public int getRateLimitFrequencyInSeconds() {
        return rateLimitFrequencyInSeconds;
    }

    public void setRateLimitFrequencyInSeconds(int rateLimitFrequencyInSeconds) {
        this.rateLimitFrequencyInSeconds = rateLimitFrequencyInSeconds;
    }

    /**
     * @return the external storage path for the task input payload
     */
    public String getExternalInputPayloadStoragePath() {
        return externalInputPayloadStoragePath;
    }

    /**
     * @param externalInputPayloadStoragePath the external storage path where the task input payload is stored
     */
    public void setExternalInputPayloadStoragePath(String externalInputPayloadStoragePath) {
        this.externalInputPayloadStoragePath = externalInputPayloadStoragePath;
    }

    /**
     * @return the external storage path for the task output payload
     */
    public String getExternalOutputPayloadStoragePath() {
        return externalOutputPayloadStoragePath;
    }

    /**
     * @param externalOutputPayloadStoragePath the external storage path where the task output payload is stored
     */
    public void setExternalOutputPayloadStoragePath(String externalOutputPayloadStoragePath) {
        this.externalOutputPayloadStoragePath = externalOutputPayloadStoragePath;
    }

    public Task copy() {
        Task copy = new Task();
        copy.setCallbackAfterSeconds(callbackAfterSeconds);
        copy.setCallbackFromWorker(callbackFromWorker);
        copy.setCorrelationId(correlationId);
        copy.setInputData(inputData);
        copy.setOutputData(outputData);
        copy.setReferenceTaskName(referenceTaskName);
        copy.setStartDelayInSeconds(startDelayInSeconds);
        copy.setTaskDefName(taskDefName);
        copy.setTaskType(taskType);
        copy.setWorkflowInstanceId(workflowInstanceId);
        copy.setWorkflowType(workflowType);
        copy.setResponseTimeoutSeconds(responseTimeoutSeconds);
        copy.setStatus(status);
        copy.setRetryCount(retryCount);
        copy.setPollCount(pollCount);
        copy.setTaskId(taskId);
        copy.setWorkflowTask(workflowTask);
        copy.setDomain(domain);
        copy.setInputMessage(inputMessage);
        copy.setOutputMessage(outputMessage);
        copy.setRateLimitPerFrequency(rateLimitPerFrequency);
        copy.setRateLimitFrequencyInSeconds(rateLimitFrequencyInSeconds);
        copy.setExternalInputPayloadStoragePath(externalInputPayloadStoragePath);
        copy.setExternalOutputPayloadStoragePath(externalOutputPayloadStoragePath);

        return copy;
    }


    @Override
    public String toString() {
        return "Task{" +
                "taskType='" + taskType + '\'' +
                ", status=" + status +
                ", inputData=" + inputData +
                ", referenceTaskName='" + referenceTaskName + '\'' +
                ", retryCount=" + retryCount +
                ", seq=" + seq +
                ", correlationId='" + correlationId + '\'' +
                ", pollCount=" + pollCount +
                ", taskDefName='" + taskDefName + '\'' +
                ", scheduledTime=" + scheduledTime +
                ", startTime=" + startTime +
                ", endTime=" + endTime +
                ", updateTime=" + updateTime +
                ", startDelayInSeconds=" + startDelayInSeconds +
                ", retriedTaskId='" + retriedTaskId + '\'' +
                ", retried=" + retried +
                ", executed=" + executed +
                ", callbackFromWorker=" + callbackFromWorker +
                ", responseTimeoutSeconds=" + responseTimeoutSeconds +
                ", workflowInstanceId='" + workflowInstanceId + '\'' +
                ", workflowType='" + workflowType + '\'' +
                ", taskId='" + taskId + '\'' +
                ", reasonForIncompletion='" + reasonForIncompletion + '\'' +
                ", callbackAfterSeconds=" + callbackAfterSeconds +
                ", workerId='" + workerId + '\'' +
                ", outputData=" + outputData +
                ", workflowTask=" + workflowTask +
                ", domain='" + domain + '\'' +
                ", inputMessage='" + inputMessage + '\'' +
                ", outputMessage='" + outputMessage + '\'' +
                ", rateLimitPerFrequency=" + rateLimitPerFrequency +
                ", rateLimitFrequencyInSeconds=" + rateLimitFrequencyInSeconds +
                ", externalInputPayloadStoragePath='" + externalInputPayloadStoragePath + '\'' +
                ", externalOutputPayloadStoragePath='" + externalOutputPayloadStoragePath + '\'' +
                '}';
    }
}
